package com.iot.mqtt;

import com.iot.compont.ServerNetwork;
import com.iot.gateway.GatewayEntity;
import com.iot.messaage.FromDeviceMessageContext;
import com.iot.network.MqttDeviceSession;
import com.iot.network.WebSocketDeviceSession;
import com.iot.test.TestDeviceMessageDecoder;
import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.mqtt.MqttEndpoint;
import io.vertx.mqtt.MqttServer;
import io.vertx.mqtt.MqttServerOptions;
import io.vertx.mqtt.messages.MqttPublishMessage;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;

/**
 * @author Administrator
 */
@Slf4j
public class MqttServerNetwork implements ServerNetwork {
    private Vertx vertx;

    @Override
    public ServerNetwork createServer(GatewayEntity gatewayEntity) {
        vertx = Vertx.vertx();
        int numberOfInstance = Math.min(Runtime.getRuntime().availableProcessors() * 2,
                Math.max(1, 1));
        List<MqttServer> instances = new ArrayList<>(numberOfInstance);
        MqttServerOptions mqttServerOptions = new MqttServerOptions();
        mqttServerOptions.setHost(gatewayEntity.getHost());
        mqttServerOptions.setPort(gatewayEntity.getProd());
        mqttServerOptions.setMaxMessageSize(8096);
        for (int i = 0; i < numberOfInstance; i++) {

            MqttServer mqttServer = MqttServer.create(vertx, mqttServerOptions);
            instances.add(mqttServer);
        }
        for (MqttServer instance : instances) {

            instance.endpointHandler(event -> {
                handleConnection(event);
            });
            vertx
                    .nettyEventLoopGroup()
                    .execute(() -> instance
                            .listen(result -> {
                                if (result.succeeded()) {
                                    System.out.println(result.result().actualPort());
                                } else {
                                }
                            }));
        }
        return this;
    }

    public void handleConnection(MqttEndpoint mqttEndpoint) {

        mqttEndpoint.accept();

        mqttEndpoint.publishHandler(message->{
            MqttPublishMessage mqttPublishMessage=message;
            String password = mqttEndpoint.auth().getPassword();
            String username = mqttEndpoint.auth().getUsername();
            System.out.println(password + username);
            Buffer payload = message.payload();
            String s = message.topicName();
            System.out.println(s);
            System.out.println(payload.toString());
            MqttDeviceSession session=new MqttDeviceSession(mqttEndpoint);
            TestDeviceMessageDecoder testDeviceMessageDecoder=new TestDeviceMessageDecoder();
            testDeviceMessageDecoder.decode(FromDeviceMessageContext.of(session));
        });
    }

    @Override
    public Runnable cloneServer() {
        return null;
    }
}
